package AppKpi

import Utile.JedisUtil
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedis

/**
  * Created by yangqiyuan on 2018/3/29.
  */
object AppKpi {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    .setAppName(AppKpi.getClass.getSimpleName)
    .setMaster("local[*]")
    val spark: SparkContext = new SparkContext(conf)

    val context: SQLContext = new SQLContext(spark)

    import context.implicits._
    val file: RDD[String] = spark.textFile("app_dict.txt")
    val parquet: DataFrame = context.read.parquet("parquet")

    val middle_t: RDD[(String)] =
    file.filter(f => f.length>=4).map(data => {
    val split = data.split("\t")
      split(4)
    })
    val appid_rules: Array[(String)] = middle_t.collect()
   val broadcast: Broadcast[Array[(String)]] = spark.broadcast(appid_rules)

    // 2.判断每个字段有没有appname,就去另一个文件找出appid
     val appRdd: RDD[(String, List[Int])] = parquet.map(p => {
       val appname: String = p.getAs("appname")
       val appid: String = p.getAs("appid")
       //val provincename: String = p.getAs("provincename")
       //val cityname: String = p.getAs("cityname")
       val requestmode: Int = p.getAs("requestmode")
       val processnode: Int = p.getAs("processnode")
       val iseffective: Int = p.getAs("iseffective")
       val isbilling: Int = p.getAs("isbilling")
       val isbid: Int = p.getAs("isbid")
       val iswin: Int = p.getAs("iswin")
       val adorderid: Int = p.getAs("adorderid")


       //对取出来的数据进行判断
       val init_request = if (requestmode == 1 && processnode >= 1) 1 else 0

       val effective_request = if (requestmode == 1 && processnode >= 2) 1 else 0

       val ad_request = if (requestmode == 1 && processnode == 3) 1 else 0

       val bid_count = if (iseffective.equals(1) && isbilling.equals(1) && isbid.equals(1) && adorderid != 0) 1 else 0

       val sucess_bid_count = if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0

       val show_count = if (requestmode == 2 && iseffective == 1) 1 else 0

       val click_count = if (requestmode == 3 && iseffective == 1) 1 else 0

       val ad_cost = if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0

       val ad_consume = if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0
       if (appname.equals("")&&appid!="") {
         val value: Array[(String)] = broadcast.value
         val rst_appid: Array[String] = value.map(x => {
           if (appid.equals(x)) {
             x
           } else {
             "unkown"
           }

         })
         val string: String = rst_appid.mkString(",")
         (string, List(init_request, effective_request, ad_request, bid_count, sucess_bid_count, show_count, click_count, ad_cost, ad_consume))
       }

       else {
         (appname, List(init_request, effective_request, ad_request, bid_count, sucess_bid_count, show_count, click_count, ad_cost, ad_consume))

       }
     })
    val appKpi: RDD[(String, List[Int])] =
     appRdd.reduceByKey((list1, list2) => {
      list1.zip(list2).map(x => x._1 + x._2)

    })

    val result: RDD[(String, String)] = appKpi.map(x => {
      (x._1, x._2.mkString(","))
    })

    result.foreach(x=>{
      val jedis: Jedis = JedisUtil.getjedis()
      jedis.hset("appname",x._1,x._2)
      jedis.close()
    })

  }

}
